%%%-------------------------------------------------------------------
%%% File    : jports.erl
%%% Author  :  <alex.arnon>
%%% Description : Library for running JPorts.
%%% Created :  1 Jan 2010 by  <alex.arnon>
%%%-------------------------------------------------------------------
-module(jports).

-behaviour(gen_server).

%% API
-export([open/1, open/3, link/1, unlink/1, close/1, close/2]).
-export([send/2, set_owner/2, info/1]).

%% Tests
-export([open_ping_pong/0,
         run_ping_pong_stress_test/0,
         run_ping_pong_stress_test/2]).


%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
         terminate/2, code_change/3]).


%% -define(DEBUG, debug).

-define(EXIT_OK, 0).
-define(EXIT_ABNORMAL_TERMINATION, 1).


-record(state, {port,                   % :: port() | exit_status().
                owner}).                % :: pid().


%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: start_link() -> {ok,Pid} | ignore | {error,Error}
%% Description: Starts the server
%%--------------------------------------------------------------------
open(CmdLine) ->
    open(CmdLine, ".", []).

open(CmdLine, _Pwd, _Env) ->
    assert_is_proplist(_Env),
    assert_is_directory(_Pwd),
    CreationArgs = [binary, eof, use_stdio, exit_status, {packet, 4}]
        ++ if _Pwd == "." -> []; true -> [{pwd, _Pwd}] end
        ++ if _Env == []  -> []; true -> [{env, _Env}] end,
    Port = erlang:open_port({spawn, CmdLine}, CreationArgs),
    Creator = self(),
    {ok,JPort} = gen_server:start_link(?MODULE, [Port, Creator], []),
    erlang:port_connect(Port, JPort),
    JPort.


link(JPort) ->
    erlang:link(JPort).


unlink(JPort) ->
    erlang:unlink(JPort).


close(JPort) ->
    gen_server:call(JPort, {close, ?EXIT_ABNORMAL_TERMINATION}).

close(JPort, ExitStatus) ->
    gen_server:call(JPort, {close, ExitStatus}).


send(JPort, Message) ->
    gen_server:call(JPort, {send_term, Message}).


set_owner(JPort, NewOwner) when is_pid(NewOwner) ->
    gen_server:call(JPort, {set_owner, NewOwner}).


info(JPort) ->
    gen_server:call(JPort, info).


%%====================================================================
%% gen_server callbacks
%%====================================================================

%%--------------------------------------------------------------------
%% Function: init(Args) -> {ok, State} |
%%                         {ok, State, Timeout} |
%%                         ignore               |
%%                         {stop, Reason}
%% Description: Initiates the server
%%--------------------------------------------------------------------
init([Port, InitialOwner]) ->
    log("init: ~w, ~w", [Port, InitialOwner]),
    {ok, #state{port = Port, owner = InitialOwner}}.

%%--------------------------------------------------------------------
%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} |
%%                                      {reply, Reply, State, Timeout} |
%%                                      {noreply, State} |
%%                                      {noreply, State, Timeout} |
%%                                      {stop, Reason, Reply, State} |
%%                                      {stop, Reason, State}
%% Description: Handling call messages
%%--------------------------------------------------------------------
handle_call({close, ExitStatus}, _From, State=#state{port=Port,owner=Owner}) ->
    if
        is_integer(Port) ->
            %% Do nothing, already exited
            log("close again: exit_status=~w, existing=~w", [Port, ExitStatus]),
            {stop, normal, already_stopped, State};
        is_port(Port) ->
            %% Close and return requested exit status to owner.
            log("close: exit_status=~w", [ExitStatus]),
            CRes = (catch erlang:port_close(Port)),
            log("res: ~w", [CRes]),
            send_exit(Owner, ExitStatus),
            {stop, normal, ok, State};
        true ->
            %% Unexpected state??
            log("close: bad jport state: port=~w", [Port]),
            {reply, {error, {bad_state, {port, Port}}}, State}
    end;


handle_call({send_term,Message}, _From, State = #state{port=Port}) ->
    case Port of
        ExitStatus when is_integer(ExitStatus) ->
            log("send term when closed: ~w", [Message]),
            {reply, {error,closed}, State};
        Port ->
            log("send term: ~w", [Message]),
            CRes = (catch erlang:port_command(Port, term_to_binary(Message))),
            log("res: ~w", [CRes]),
            {reply, ok, State}
    end;


handle_call({set_owner,NewOwner}, _From, State=#state{port=Port,owner=Owner}) ->
    case Owner of
        NewOwner ->
            %% New is same as old - do nothing.
            log("set owner - unchanged"),
            {reply, ok, State};
        _ ->
            %% Ensure new owner knows the port has closed.
            log("set new owner: ~w", [NewOwner]),
            case Port of
                ExitStatus when is_integer(ExitStatus) ->
                    send_exit(NewOwner, ExitStatus);
                _Otherwise ->
                    nop
            end,
            {reply, ok, State#state{owner=NewOwner}}
    end;


handle_call(info, _From, State=#state{port=Port,owner=Owner}) ->
    {reply,
     [{is_closed, not is_port(Port)},
      {port, if is_port(Port) -> Port ; true -> {exit_status,Port} end},
      {owner, Owner},
      {port_info, if is_port(Port) -> erlang:port_info(Port);
                     true          -> not_available
                  end}
     ],
     State};


handle_call(_Request, _From, State) ->
    log("unexpected call: req=~w from=~w", [_Request, _From]),
    Reply = ok,
    {reply, Reply, State}.

%%--------------------------------------------------------------------
%% Function: handle_cast(Msg, State) -> {noreply, State} |
%%                                      {noreply, State, Timeout} |
%%                                      {stop, Reason, State}
%% Description: Handling cast messages
%%--------------------------------------------------------------------
handle_cast(_Msg, State) ->
    log("unexpected cast: msg=~w", [_Msg]),
    {noreply, State}.

%%--------------------------------------------------------------------
%% Function: handle_info(Info, State) -> {noreply, State} |
%%                                       {noreply, State, Timeout} |
%%                                       {stop, Reason, State}
%% Description: Handling all non call/cast messages
%%--------------------------------------------------------------------
handle_info({Port,{data,Data}}, State=#state{port=Port,owner=Owner}) ->
    log("got data: ~w", [Data]),
    Term = binary_to_term(Data),
    log("which is term: ~w", [Term]),
    Owner ! {jport_data, self(), Term},
    {noreply, State};

handle_info({Port,eof}, State=#state{port=Port}) ->
    %% Port stream has closed - exit status is the real stop signal for us.
    log("got eof", []),
    {noreply, State};

handle_info({Port,{exit_status,ExitStatus}}, State=#state{port=Port,
                                                          owner=Owner}) ->
    log("got exit: status=~w", [ExitStatus]),
    CRes = (catch send_exit(Owner, ExitStatus)),
    log("res=~w", [CRes]),
    {noreply, State#state{port=ExitStatus}};
    
handle_info(_Info, State) ->
    {noreply, State}.

%%--------------------------------------------------------------------
%% Function: terminate(Reason, State) -> void()
%% Description: This function is called by a gen_server when it is about to
%% terminate. It should be the opposite of Module:init/1 and do any necessary
%% cleaning up. When it returns, the gen_server terminates with Reason.
%% The return value is ignored.
%%--------------------------------------------------------------------
terminate(Reason, State) ->
    log("terminating: reason=~w state=~w", [Reason, State]),
    ok.

%%--------------------------------------------------------------------
%% Func: code_change(OldVsn, State, Extra) -> {ok, NewState}
%% Description: Convert process state when code is changed
%%--------------------------------------------------------------------
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%--------------------------------------------------------------------
%%% Internal functions
%%--------------------------------------------------------------------

assert_is_proplist([]) ->
    true;
assert_is_proplist([{K,V}|T]) ->
    assert_is_ascii(K),
    assert_is_ascii(V),
    assert_is_proplist(T);
assert_is_proplist(_) ->
    erlang:error({badarg,proplist_expected}).

assert_is_directory(Pwd) ->
    case filelib:is_dir(Pwd) of
        true ->
            true;
        false ->
            erlang:error({badarg,directory_expected})
    end.

assert_is_ascii(Str) when is_list(Str) ->
    case lists:all(fun(C) -> (C >= 33) andalso (C =< 127) end, Str) of
        true ->
            true;
        false ->
            erlang:error({badarg,ascii_string_expected})
    end.
                           
send_exit(Owner, ExitStatus) ->
    log("sending exit: owner=~w exit_status=~w", [Owner, ExitStatus]),
    Owner ! {jports_exit, self(), ExitStatus}.


log(S) ->
    log(S, []).

-ifdef(DEBUG).
log(F,A) ->
    io:format("jport[~w]: "++F++"~n", [self()]++A).
-else.
log(_F,_A) ->
    ok.
-endif.


%%--------------------------------------------------------------------
%%% Tests.
%%--------------------------------------------------------------------


open_ping_pong() ->
    open("java -jar ping_pong_test.jar", ".", [{"JPORTS_LOG_LEVEL", "INFO"}]).


run_ping_pong_stress_test() ->
    run_ping_pong_stress_test(100, lists:seq(1,100)).

run_ping_pong_stress_test(NPings, StaticPayload) when NPings > 0 ->
    log("opening ping pong port"),
    P1 = open_ping_pong(),
    log("checking open ok"),
    true = is_pid(P1),
    log("linking"),
    erlang:link(P1),
    log("running test with ~w iterations and static payload = ~w",
        [NPings, StaticPayload]),
    TStart = erlang:now(),
    {TEnd, TFirst} = ping_pong_loop(P1, NPings, StaticPayload, never),
    log("finished test: ~w..~w..~w", [TStart, TFirst, TEnd]),
    log("unlinking and closing port"),
    erlang:unlink(P1),
    jports:close(P1),
    receive {jports_exit, P1, ExitStatus} -> ok end,
    log("got exit status ~w", [ExitStatus]),
    %% Stats
    TStartMs = to_ms(TStart),
    TFirstMs = to_ms(TFirst),
    TEndMs = to_ms(TEnd),
    MsStartToFirst = TFirstMs - TStartMs,
    MsStartToEnd = TEndMs - TStartMs,
    MsFirstToEnd = TEndMs - TFirstMs,
    {ok, [{npings, NPings},
          {payload, StaticPayload},
          {exit_status, ExitStatus},
          {ms_start_to_first, MsStartToFirst},
          {ms_first_to_end, MsFirstToEnd},
          {ms_start_to_end, MsStartToEnd},
          {t_start, TStart},
          {t_first, TFirst},
          {t_end, TEnd},
          {t_start_ms, TStartMs},
          {t_first_ms, TFirstMs},
          {t_end_ms, TEndMs}
         ]}.


to_ms({MSec, Sec, USec}) ->
    ((MSec*1000000) + Sec)*1000 + (USec div 1000).


ping_pong_loop(Port, 0, StaticPayload, never) ->
    ping_pong_loop(Port, 0, StaticPayload, drain_port(Port, never));
ping_pong_loop(Port, 0, _, TFirst) ->
    FinalTFirst = drain_port(Port, TFirst),
    {erlang:now(), FinalTFirst};
ping_pong_loop(Port, NPings, StaticPayload, TFirst)
  when NPings > 0 ->
    jports:send(Port, {ping, StaticPayload}),
    ping_pong_loop(Port, NPings-1, StaticPayload, drain_port(Port, TFirst)).


drain_port(Port, never) ->
    receive {jport_data, Port, {pong, _}} -> drain_port(Port, erlang:now())
    after 0 -> never
    end;
drain_port(Port, T) ->
    receive {jport_data, Port, {pong, _}} -> drain_port(Port, T)
    after 0 -> T
    end.
    
            
            
